package com.gitee.ixtf.vertx.rabbitmq.guice

import com.gitee.ixtf.core.Jassert
import com.google.inject.Singleton
import jakarta.inject.Inject
import java.time.Duration
import kotlinx.coroutines.reactor.awaitSingle
import org.reactivestreams.Publisher
import reactor.rabbitmq.ExceptionHandlers
import reactor.rabbitmq.ExceptionHandlers.RetrySendingExceptionHandler
import reactor.rabbitmq.OutboundMessage
import reactor.rabbitmq.SendOptions
import reactor.rabbitmq.Sender
import reactor.util.retry.Retry

@Singleton
class RabbitMQSender @Inject constructor(private val sender: Sender) {
  private val options =
      SendOptions()
          .trackReturned(true)
          .exceptionHandler(
              RetrySendingExceptionHandler(
                  Duration.ofHours(1),
                  Duration.ofMinutes(5),
                  ExceptionHandlers.CONNECTION_RECOVERY_PREDICATE))

  suspend fun send(message: Publisher<OutboundMessage>) {
    val errors =
        sender
            .sendWithPublishConfirms(message, options)
            .retryWhen(Retry.backoff(3, Duration.ofSeconds(3)))
            // todo check isReturned 没有路由
            .filter { !it.isAck }
            .collectList()
            .awaitSingle()
    Jassert.isEmpty(errors)
  }
}
